package com.study.flink.java.day06_exactly;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

/**
 * 自定义redis sink
 * 数据格式自由组合放到sink里面
 */
public class MyRedisSink extends RichSinkFunction<Tuple3<String, String, String>> {

    private transient Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        ParameterTool parameterTool = (ParameterTool)getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
        jedis = new Jedis(
                parameterTool.get("redis.host", "localhost"),
                parameterTool.getInt("redis.port", 6379),
                parameterTool.getInt("redis.timeout", 5000));
        jedis.auth(parameterTool.get("redis.password", "redis123456"));
        jedis.select(parameterTool.getInt("redis.db", 0));
    }

    @Override
    public void invoke(Tuple3<String, String, String> value, Context context) throws Exception {
        if(!jedis.isConnected()) {
            jedis.connect();
        }
        jedis.hset(value.f0, value.f1, value.f2);
    }

    @Override
    public void close() throws Exception {
        super.close();
        jedis.close();
    }


}
